package com.voice.client;

import com.voice.client.auth.AuthHandler;
import com.voice.client.auth.AuthenticationFactory;
import com.voice.client.channel.tcp.TCPServer;
import com.voice.client.message.entity.Message;
import com.voice.core.DefaultClient;
import com.voice.core.entity.HttpHeader;
import com.voice.core.channel.tcp.impl.IOSelectorConnectionManager;
import com.voice.core.connection.Connection;
import com.voice.core.connection.ConnectionListener;
import com.voice.core.connection.ConnectionManager;
import com.voice.core.connection.ConnectionStatus;
import com.voice.core.connection.callback.AbstractStreamListener;
import com.voice.core.connection.entity.StreamData;
import com.voice.core.exception.ClientException;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
public abstract class AbstractClientImpl extends AbstractStreamListener implements Client, ConnectionListener {

    private Profile profile;
    private ConnectionManager connectionManager;
    protected AtomicBoolean isConnected;
    private DefaultClient client;
    private AuthHandler authHandler;
    private Connection connection;
    protected TCPServer server;

    public AbstractClientImpl(Profile profile) {
        this.isConnected = new AtomicBoolean(false);
        this.client = new DefaultClient();
        this.profile = profile;
        this.authHandler = AuthenticationFactory.getAuthHandler(profile);
    }

    private CompletableFuture<StreamData> sendAuth(Connection connection) {
        HttpHeader headers = new HttpHeader();
        authHandler.getAuthParams().forEach((key,value)->{
            headers.put(key,value);
        });
        headers.path(path());
        return this.client.sendRequest(connection, headers,null);
    }

    protected Message convertStreamData2Message(StreamData streamData) {
        if (!streamData.isOk()) {
            throw new ClientException("status is not success, code: " + streamData.getHeaders().code() + ", content: " + new String(streamData.readAllData()));
        } else {
            byte[] payload = streamData.readAllData();
            int messageId = streamData.getHeaders().get("x-message-id",Integer.class);
            String topic = null;
            if (streamData.getHeaders().contains("x-topic")) {
                topic = streamData.getHeaders().get("x-topic",String.class);
            }
            int qos = 0;
            if (streamData.getHeaders().contains("x-qos")) {
                qos = streamData.getHeaders().get("x-qos",Integer.class);
            }
            long generateTime = 0L;
            if (streamData.getHeaders().contains("x-generate-time")) {
                generateTime = streamData.getHeaders().get("x-generate-time",Long.class);
            }
            return new Message(payload, topic, messageId, qos, generateTime);
        }
    }

    protected void doConnect() {
        try {
            if (this.connectionManager == null) this.connectionManager = new IOSelectorConnectionManager();
            connection=connectionManager.connect(new InetSocketAddress(profile.getHost(),profile.getPort()));
            connection.setDefaultStreamListener(this);
            CompletableFuture<StreamData> completableFuture=this.sendAuth(connection);
            StreamData streamData = completableFuture.get(15L, TimeUnit.SECONDS);
            log.info("{}",streamData);
            if (!streamData.isOk()) {
                throw new ClientException("connect to server failed", streamData);
            } else {
                log.info("connect to server success!");
                connection.setStatus(ConnectionStatus.AUTHORIZED);
                connection.setConnectionListener(this);
                connection.setDefaultStreamListener(this);
                this.isConnected.set(true);
            }
        } catch (ExecutionException e) {
            throw new ClientException(e.getCause());
        } catch (TimeoutException | InterruptedException e) {
            throw new ClientException(e);
        }catch (IOException e) {
            throw new ClientException(e);
        }
    }

    protected void checkStarted() {
        if (!this.isConnected.get()) {
            throw new ClientException("client is not connected, please connect first");
        }
    }

    @Override
    public void disconnect() {
        this.checkStarted();
        if (this.isConnected.compareAndSet(true, false)) {
            this.connectionManager.shutdown();
            this.server.close();
        }
    }

    @Override
    public boolean isConnected() {
        return this.isConnected.get();
    }

    @Override
    public void onStatusChange(ConnectionStatus status, Connection connection) {
        if(status==ConnectionStatus.CLOSED){
            this.client = null;
            this.isConnected.set(false);
            if(this.connectionManager!=null) this.connectionManager.shutdown();
            if(this.server!=null) this.server.close();
            log.info("connect is closed!!!");
        }
    }

    public abstract String path();

    public DefaultClient client(){
        return client;
    }

    public Connection connection(){
        return connection;
    }
}
